package center_rpc

import (
	"common/connect_mgr"
	"common/pb"
	rs2 "common/rs"
	"common/uuid"
	"context"
	"fmt"
	mapset "github.com/deckarep/golang-set/v2"
	"time"
)

//向中心节点RPC请求

type BeginSegmentRespone struct {
	Nodes    []NodeInfo
	StreamId []byte
}

type BeginObjectParam struct {
	PlainObjectKey      string
	EncryptedObjectKey  []byte
	EncryptedContentKey []byte
	Bucket              string
	ExpireTime          int64
	InlineData          []byte
	Attributes          map[string]string
}

func (r *BeginObjectParam) String() string {
	return fmt.Sprintf("Bucket %s", r.Bucket)
}

// 请求获取节点
func (cRpc *CenterRpc) BeginObject(ctx context.Context, req BeginObjectParam) (*pb.ObjectBeginResponse, error) {
	client, connClose, err := cRpc.getDrpcClient(ctx)
	if err != nil {
		return nil, err
	}
	defer connClose()
	// set a deadline for the operation
	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

	fmt.Println("上传object ", req.String())
	ctx = context.Background()
	return client.BeginObject(ctx, &pb.ObjectBeginRequest{
		Header:              cRpc.Header(),
		Bucket:              []byte(req.Bucket),
		EncryptedObjectKey:  req.EncryptedObjectKey,
		EncryptedContentKey: req.EncryptedContentKey,
		Version:             0,
		PlainObjectKey:      []byte(req.PlainObjectKey),
		InlineData:          req.InlineData,
		Attributes:          req.Attributes,
	})
}

type ListObjectOpts struct {
	EncryptedPrefix []byte
	Cursor          []byte
	Recursive       bool
	Limit           int32
}

// 请求获取节点
func (cRpc *CenterRpc) ListObjects(ctx context.Context, bucket string, opts ListObjectOpts) ([]*pb.ObjectListItem, error) {
	client, connClose, err := cRpc.getDrpcClient(ctx)
	if err != nil {
		return nil, err
	}
	defer connClose()
	// set a deadline for the operation
	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

	clientResponse, err := client.ListObjects(ctx, &pb.ObjectListRequest{
		Header:          cRpc.Header(),
		Bucket:          []byte(bucket),
		EncryptedPrefix: opts.EncryptedPrefix,
		Cursor:          opts.Cursor,
		Recursive:       opts.Recursive,
		Limit:           opts.Limit,
	})
	if err != nil {
		return nil, err
	}
	return clientResponse.Items, nil
}

func (cRpc *CenterRpc) UpdateNode(ctx context.Context, update int32, include [][]byte, exclude [][]byte) []connect_mgr.Device {
	infos := make([]connect_mgr.Device, 0)
	client, connClose, err := cRpc.getDrpcClient(ctx)
	if err != nil {
		return infos
	}
	defer connClose()
	//ctx, cancel := context.WithTimeout(ctx, time.Second*10)
	//defer cancel()

	clientResponse, err := client.UploadNodeSelection(ctx, &pb.UploadNodeSelectionRequest{
		Header:        cRpc.Header(),
		IncludeNodeId: include,
		ExcludeNodeId: exclude,
		Quantity:      update,
	})
	if err != nil {
		return infos
	}

	for _, info := range clientResponse.NodeInfo {
		nodeId, _ := uuid.FromBytes(info.NodeId)

		infos = append(infos, connect_mgr.Device{
			Id:     info.StorageNodeAddress.DeviceId,
			Port:   uint16(info.StorageNodeAddress.Port),
			NodeId: nodeId,
		})
	}
	return infos
}
func (cRpc *CenterRpc) GetNodeByDevice(ctx context.Context, deviceId string) (*connect_mgr.Device, error) {
	client, connClose, err := cRpc.getDrpcClient(ctx)
	if err != nil {
		return nil, err
	}
	defer connClose()
	clientResponse, err := client.GetNodeWithDevice(ctx, &pb.GetNodeWithDeviceRequest{
		Header: cRpc.Header(),
		DevId:  []byte(deviceId),
	})
	if err != nil {
		return nil, err
	}
	info := clientResponse.NodeInfo
	nodeId, _ := uuid.FromBytes(info.NodeId)

	return &connect_mgr.Device{
		Id:     info.StorageNodeAddress.DeviceId,
		Port:   uint16(info.StorageNodeAddress.Port),
		NodeId: nodeId,
	}, nil
}

// 请求获取节点
func (cRpc *CenterRpc) BeginSegment(ctx context.Context, index int, streamId []byte, nodeLimit int32, lastGoodNode []uuid.UUID) (*BeginSegmentRespone, error) {
	var response = &BeginSegmentRespone{
		Nodes:    make([]NodeInfo, 0),
		StreamId: streamId,
	}
	client, connClose, err := cRpc.getDrpcClient(ctx)
	if err != nil {
		return response, err
	}
	defer connClose()
	// set a deadline for the operation
	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

	oldNodeSet := mapset.NewSet[uuid.UUID]()

	includes := make([][]byte, 0)
	for i := range lastGoodNode {
		includes = append(includes, lastGoodNode[i][:])
		oldNodeSet.Add(lastGoodNode[i])
	}

	clientResponse, err := client.BeginSegment(ctx, &pb.SegmentBeginRequest{
		Header: cRpc.Header(),
		Position: &pb.SegmentPosition{
			Index: int32(index),
		},
		MaxOrderLimit: 0,
		Quantity:      nodeLimit,
		StreamId:      streamId,
		IncludeNodeId: includes,
	})
	if err != nil {
		fmt.Println("call BeginSegment err: ", err)
		return response, err
	}

	response.StreamId = clientResponse.SegmentId

	newNodeSet := mapset.NewSet[uuid.UUID]()

	//newNodeIds := make([]uuid.UUID, 0)
	pieceNum := 0
	for _, limit := range clientResponse.NodeInfo {
		if limit.StorageNodeAddress.Port <= 0 || limit.StorageNodeAddress.Port > 0xffff {
			fmt.Printf("error limit: nid %x devid %s port %d\n", limit.NodeId,
				limit.StorageNodeAddress.DeviceId, limit.StorageNodeAddress.Port)
			continue
		}
		_nodeId, _ := uuid.FromBytes(limit.NodeId)
		newNodeSet.Add(_nodeId)

		response.Nodes = append(response.Nodes, NodeInfo{
			DevId: limit.StorageNodeAddress.DeviceId,
			//LocalDeviceId: "ZKGJ3GDKUPXS6HCR71A0",
			Port:   uint16(limit.StorageNodeAddress.Port),
			Ip:     limit.StorageNodeAddress.Ip,
			Ipv6:   limit.StorageNodeAddress.Ipv6,
			NodeId: _nodeId,
		})
		pieceNum++
	}
	//打印新的节点有多少
	//diffSet := oldNodeSet.Difference(newNodeSet)
	//fmt.Printf("更新节点: %d 个, 新节点数去重 %d\n", diffSet.Cardinality(), newNodeSet.Cardinality())

	return response, nil
}

func (cRpc *CenterRpc) CommitSegment(ctx context.Context, streamId []byte, padding int,
	segmentIndex int32,
	dataLen int,
	result []*pb.PieceStoreInfo) error {
	client, connClose, err := cRpc.getDrpcClient(ctx)
	if err != nil {
		return err
	}
	defer connClose()
	_, err = client.CommitSegment(ctx, &pb.SegmentCommitRequest{
		Header:       cRpc.Header(),
		StreamId:     streamId,
		SegmentIndex: segmentIndex,
		Infos:        result,
		Padding:      int32(padding),
		SegmentSize:  int32(dataLen),
	})
	if err != nil {
		return err
	}

	return nil
}
func (cRpc *CenterRpc) CommitObject(ctx context.Context, streamid, encObjectKey []byte, segmentSize int, segmentCount int, plainSize int64, rs rs2.Rs) error {

	client, connClose, err := cRpc.getDrpcClient(ctx)
	if err != nil {
		return err
	}
	defer connClose()

	_, err = client.CommitObject(ctx, &pb.ObjectCommitRequest{
		Header:       cRpc.Header(),
		StreamId:     streamid,
		PlainSize:    plainSize,
		SegmentCount: int32(segmentCount),
		SegmentSize:  int32(segmentSize),
		Rs: &pb.RS{
			Total:     rs.Total,
			MinUpload: rs.MinUpload,
			Repair:    rs.Repair,
			Required:  rs.Required,
		},
		EncryptedObjectKey: encObjectKey,
	})
	return err
}
